package com.hzgj.bcl.soa.rpc;

import com.hzgj.bcl.concurrent.FutureListener;
import com.hzgj.bcl.soa.ChainStack;
import com.hzgj.bcl.soa.Event;
import com.hzgj.bcl.soa.Interceptor;
import com.hzgj.bcl.soa.rpc.client.RequestOptions;
import com.hzgj.bcl.soa.rpc.client.Response;
import com.hzgj.bcl.soa.rpc.client.RpcRequest;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.soa.util.NotifyingFuture;
import com.hzgj.bcl.soa.util.Rsp;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public abstract class MessageDispatcher implements AsyncRequestHandler, Closeable {
    protected final Logger logger = LoggerFactory.getLogger(getClass());
    protected ChainStack stack;
    protected RequestCorrelator corr;
    protected RequestHandler req_handler;
    protected boolean async_dispatching;

    public MessageDispatcher(ChainStack stack) {
        this.stack = stack;
    }

    public MessageDispatcher(ChainStack stack, RequestHandler req_handler) {
        this(stack);
        setRequestHandler(req_handler);
    }

    public boolean asyncDispatching() {
        return async_dispatching;
    }

    public MessageDispatcher asyncDispatching(boolean flag) {
        async_dispatching = flag;
        if (corr != null) {
            corr.asyncDispatching(flag);
        }
        return this;
    }

    public void start() throws Exception {
        stack.insertInterceptorAtTop(new InterceptorAdapter());
        if (corr == null) {
            corr = new RequestCorrelator(stack, this).asyncDispatching(async_dispatching);
        }
        stack.start();
        corr.start();
    }

    @Override
    public void close() throws IOException {
        stop();
    }

    public void stop() {
        if (corr != null) {
            corr.stop();
        }
        stack.stop();
    }

    public final void setRequestHandler(RequestHandler rh) {
        req_handler = rh;
    }

    public void done(long req_id) {
        corr.done(req_id);
    }

    public <T> T sendMessage(Message msg, RequestOptions opts) throws Exception {
        RpcRequest<T> req = new RpcRequest<>(msg, corr, opts);
//            req.setBlockForResults(false);
        req.execute();
        Rsp<T> rsp = req.getResult();
        Throwable exception = rsp.getException();
        if (exception != null) {
            if (exception instanceof Error) {
                throw (Error) exception;
            } else if (exception instanceof RuntimeException) {
                throw (RuntimeException) exception;
            } else if (exception instanceof Exception) {
                throw (Exception) exception;
            } else if (exception instanceof ConnectException) {
                throw (ConnectException) exception;
            } else if (exception instanceof TimeoutException) {
                //TODO：from what
                throw new TimeoutException("timeout waiting for response from " + ", request: " + req.toString());
            } else {
                throw new RuntimeException(exception);
            }
        }
        return rsp.getValue();
    }

    public <T> NotifyingFuture<T> sendMessageWithFuture(Message msg, RequestOptions options, FutureListener<T> listener) throws Exception {
        RpcRequest<T> req = new RpcRequest<>(msg, corr, options);
        if (listener != null) {
            req.setListener(listener);
        }
        req.setBlockForResults(false);
        req.execute();
        if (options != null && options.getReferenceDef().isAsync()) {
            return new NullFuture<>(null);
        }
        return req;
    }

    @Override
    public Object handle(Message msg) throws Exception {
        if (req_handler != null) {
            return req_handler.handle(msg);
        }
        return null;
    }

    @Override
    public void handle(Message request, Response response) throws Exception {
        if (req_handler != null) {
            if (req_handler instanceof AsyncRequestHandler) {
                ((AsyncRequestHandler) req_handler).handle(request, response);
            } else {
                Object retval = req_handler.handle(request);
                if (response != null) {
                    response.send(retval, (String) request.getHeader(Constants.MESSAGE_HEADER_PROTOCOL), false);
                }
            }
            return;
        }

        Object retval = handle(request);
        if (response != null) {
            response.send(retval, (String) request.getHeader(Constants.MESSAGE_HEADER_PROTOCOL), false);
        }
    }

    class InterceptorAdapter extends Interceptor {

        @Override
        public String getName() {
            return "MessageDispatcher";
        }

        @Override
        public Object up(Event evt) {
            if (corr != null) {
                corr.receive(evt);
            }
            return null;
        }

        @Override
        public Object down(Event evt) {
            return down_int.down(evt);
        }
    }
}
